package com.sunzm.flink.hotitems

import java.time.Duration

import com.sunzm.flink.hotitems.bean.{ItemViewCount, UserBehavior}
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 实时热门商品统计:
 * 每隔5分钟统计一次最近一小时每个商品的点击量
 */
object HotItemsJob {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true
  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //为了便于观察，并行度设置为1，实际生产环境不能设置成1
    env.setParallelism(1)

    //读取数据
    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)
    //val dataStream: DataStream[String] = env.readTextFile("data/flink/case-demo/UserBehavior.csv")

    //创建一个Watermark生成策略对象，并提取事件时间
    val watermarkStrategy = WatermarkStrategy
      //数据没有乱序，WaterMark不需要设置乱序程度
      .forBoundedOutOfOrderness[UserBehavior](Duration.ofSeconds(0))
      .withTimestampAssigner(new SerializableTimestampAssigner[UserBehavior] {
        override def extractTimestamp(element: UserBehavior, recordTimestamp: Long): Long = {
          element.timestamp * 1000L
        }
      })

    val mapedDataStream: DataStream[UserBehavior] = dataStream.map(line => {
      val fields = line.split(",")

      UserBehavior(fields(0).toLong, fields(1).toLong, fields(2).toInt, fields(3), fields(4).toLong)
    })
      // 指定WaterMark生成策略，以及提取事件时间
      .assignTimestampsAndWatermarks(watermarkStrategy)

    //需求：每隔5分钟统计一次最近一小时每个商品的点击量（用PV来计算）
    //过滤出PV类型的数据
    val pvDataStream: DataStream[UserBehavior] = mapedDataStream.filter(userBehavior => StringUtils.equals("pv", userBehavior.behavior))

    val itemClountDS: DataStream[ItemViewCount] = pvDataStream
      //按照商品ID分组，因为需要统计每个商品的点击量
      .keyBy(_.itemId)
      //开一个大小为1小时，滑动步长为5分钟，基于事件时间的滑动窗口
      .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
      // 这种是把所有的数据攒到一个窗口，等窗口关闭的时候，统一处理
      //.process()
      //这种是增量聚合，来一条数据就聚合一次，最后窗口关闭的时候，再整体聚合一次输出
      .aggregate(new ItemCountAgg, new WindowCountResult)


    itemClountDS.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  /**
   * 自定义预聚合函数类，每来一个数据就count加1
   */
  private class ItemCountAgg extends  AggregateFunction[UserBehavior, Int, Int]{
    //创建一个用于保存中间聚合结果的累加器，这里进行初始化
    override def createAccumulator(): Int = 0

    //来一条数据，就应该+1
    override def add(value: UserBehavior, accumulator: Int): Int = {
      accumulator + 1
    }

    //获取结果
    override def getResult(accumulator: Int): Int = accumulator

    override def merge(a: Int, b: Int): Int = a + b
  }

  /**
   * 窗口函数
   */
  private class WindowCountResult extends WindowFunction[Int, ItemViewCount, Long, TimeWindow]{
    override def apply(key: Long, window: TimeWindow, input: Iterable[Int], out: Collector[ItemViewCount]): Unit = {

      //商品ID
      val itemId = key

      //获取点击数量
      /*var count = 0
      input.foreach(ct => {
        count += ct
      })*/
      //因为我们做过预聚合，这里得到的数据只有一条
      val count = input.iterator.next()

      //获取窗口开始时间和结束时间
      val windowStart = window.getStart
      val windowEnd = window.getEnd

      //时间格式化
      val fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")

      val windowStartStr = fastDateFormat.format(windowStart)
      val windowEndStr = fastDateFormat.format(windowEnd)

      //输出结果
      out.collect(ItemViewCount(windowStartStr, windowEndStr, itemId, count))
    }
  }
}
